package data_manipulate;

import Utils.FileUtil;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.util.Order;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.*;

/*
data are the same with TopN1
orderid userid payment productid
1,9819,100,121
...
 */
public class TopN2 {

    public static String inputPath = "hdfs://localhost:9000/top/top_payment";
    public static String outputPath = "hdfs://localhost:9000/top/output_payment2";

    // this time, we only store top 5 payment in map
    // due to distributed nature, each map will return 5 to oen reducer
    // thus the total number of data needs processing is map num * 5 for reducer.
    public static void main(String[] args)
            throws IOException, InterruptedException, ClassNotFoundException {
        Configuration conf = new Configuration();
        // we need to store topN value into configuration
        conf.setInt("top", 5);
        FileUtil fileUtil = new FileUtil(conf, outputPath);
        fileUtil.doError(inputPath);
        fileUtil.doDelete(outputPath);

        Job job = Job.getInstance(conf, "topN2");
        job.setJarByClass(TopN2.class);
        // for top problem, we only need on reducer process.
        job.setNumReduceTasks(1);

        job.setMapperClass(doMapper.class);
        job.setMapOutputKeyClass(OrderBean.class);
        job.setMapOutputValueClass(NullWritable.class);

        job.setReducerClass(doReducer.class);
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(Text.class);

        //job.setSortComparatorClass(doComparator.class);

        FileInputFormat.setInputPaths(job, new Path(inputPath));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }

    public static class OrderBean implements WritableComparable<OrderBean> {

        private int orderId;
        private int userId;
        private double payment;
        private int productId;

        public OrderBean() {
        }

        public int getOrderId() {
            return orderId;
        }

        public void setOrderId(int orderId) {
            this.orderId = orderId;
        }

        public int getUserId() {
            return userId;
        }

        public void setUserId(int userId) {
            this.userId = userId;
        }

        public double getPayment() {
            return payment;
        }

        public void setPayment(double payment) {
            this.payment = payment;
        }

        public int getProductId() {
            return productId;
        }

        public void setProductId(int productId) {
            this.productId = productId;
        }

        @Override
        public int compareTo(OrderBean o) {
            if (o == null) {
                return 1;
            } else {
                return (int) (this.payment - o.payment);
            }
        }

        @Override
        public void write(DataOutput out) throws IOException {
            //orderid userid payment productid
            out.writeInt(this.orderId);
            out.writeInt(this.userId);
            out.writeDouble(this.payment);
            out.writeInt(this.productId);
        }

        @Override
        public void readFields(DataInput in) throws IOException {
            //orderid userid payment productid
            this.orderId = in.readInt();
            this.userId = in.readInt();
            this.payment = in.readDouble();
            this.productId = in.readInt();

        }

        @Override
        public String toString() {
            return this.orderId + " " + this.userId + " " + this.payment + " " + this.productId;
        }
    }

    public static class OrderBeanComparator implements Comparator<OrderBean> {
        @Override
        public int compare(OrderBean o1, OrderBean o2) {
            if (o1 == null) {
                return o2 == null ? 0 : -1;
            } else if (o2 == null) {
                return 1;
            }
            return o1.compareTo(o2);
        }
    }

    public static class doMapper extends Mapper<LongWritable, Text, OrderBean, NullWritable> {

        private OrderBean[] beanList;
        private int top;

        @Override
        protected void setup(Mapper<LongWritable, Text, OrderBean, NullWritable>.Context context) throws IOException, InterruptedException {
            top = context.getConfiguration().getInt("top", 10);
            beanList = new OrderBean[top + 1];
        }

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, OrderBean, NullWritable>.Context context) throws IOException, InterruptedException {
            String[] splits = value.toString().trim().split(",");
            OrderBean currBean = new OrderBean();
            // packaging ...
            // orderid userid payment productid
            if (splits.length == 4) {
                currBean.setOrderId(Integer.parseInt(splits[0]));
                currBean.setUserId(Integer.parseInt(splits[1]));
                currBean.setPayment(Double.parseDouble(splits[2]));
                currBean.setProductId(Integer.parseInt(splits[3]));
                //System.out.println(bean.toString());
                add(currBean);
            }
        }

        private void add(OrderBean bean) {
            beanList[0] = bean;
            Arrays.sort(beanList, new OrderBeanComparator());
        }

        @Override
        protected void cleanup(Mapper<LongWritable, Text, OrderBean, NullWritable>.Context context) throws IOException, InterruptedException {
            for (int i = 0; i < top; i++) {
                System.out.println(i + ": " + beanList[i].toString());
                context.write(beanList[i], NullWritable.get());
            }
        }
    }

    public static class doReducer extends Reducer<OrderBean, NullWritable, IntWritable, Text> {

        private OrderBean[] beanList;
        private int top;

        private IntWritable id = new IntWritable();
        private Text result = new Text();

        @Override
        protected void setup(Reducer<OrderBean, NullWritable, IntWritable, Text>.Context context) throws IOException, InterruptedException {
            top = context.getConfiguration().getInt("top", 10);
            beanList = new OrderBean[top + 1];
        }

        @Override
        protected void reduce(OrderBean key, Iterable<NullWritable> values,
                              Reducer<OrderBean, NullWritable, IntWritable, Text>.Context context)
                throws IOException, InterruptedException {
            OrderBean currBean = new OrderBean();
            try {
                BeanUtils.copyProperties(currBean, key);
                add(currBean);
            } catch (IllegalAccessException | InvocationTargetException e) {
                e.printStackTrace();
            }

        }

        private void add(OrderBean bean) {
            beanList[0] = bean;
            Arrays.sort(beanList, new OrderBeanComparator());
        }

        @Override
        protected void cleanup(Reducer<OrderBean, NullWritable, IntWritable, Text>.Context context)
                throws IOException, InterruptedException {
            for (int i = top; i > 0; i--) {
                // orderid userid payment productid
                id.set(top - i + 1);
                result.set(beanList[i].toString());
                context.write(id, result);
            }
        }
    }

    public static class doComparator extends WritableComparator {

        protected doComparator() {
            super(OrderBean.class, true);
        }

        @Override
        public int compare(WritableComparable a, WritableComparable b) {
            OrderBean bean1 = (OrderBean) a;
            OrderBean bean2 = (OrderBean) b;
            return bean1.compareTo(bean2);
        }
    }
}
